package com.atguigu.app.dwd.db;

import com.atguigu.utils.KafkaUtil;
import com.atguigu.utils.MySqlUtil;
import com.atguigu.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

//交易域下单事务事实表
/*
主表是订单明细表（最细粒度）
join 订单表
left join 活动表
left join 优惠券表
lookup join 字典表(要求主流建表时设置处理时间字段)

（left join会产生左null，等右边数据来了，就撤回变成左右，所以要用upsert kafka，并且构建动态表连接kafka的时候要指定主键）

 */
//todo 3.创建动态表连接kafka topic_db
//todo 4.过滤出订单明细表
//todo 5.过滤出订单表
//todo 6.过滤出活动表
//todo 7.过滤出优惠卷表
//todo 8.创建动态表base_dic连接mysql的base_dic表
//todo 9.五表关联
//todo 10.创建动态表dwd_order_detail连接kafka dwd_trade_order_detail主题
//todo 11.将关联查询后的结果写入dwd_order_detail（即写入kafka dwd_trade_order_detail主题）
public class DwdTradeOrderDetail {
    public static void main(String[] args) {
        //todo 1.创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        //todo 设置状态后端

        /*
        {
    "database": "gmall",
    "table": "user_info",
    "type": "update",
    "ts": 1592227344,
    "xid": 1194519,
    "xoffset": 73,
    "data": {
        "id": 348,
        "login_name": "q3t9zd",
        "nick_name": "滢馥",
        "passwd": null,
        "name": "时荷丹",
        "phone_num": "13832212286",
        "email": "q3t9zd@live.com",
        "head_img": null,
        "user_level": "1",
        "birthday": "1988-05-11",
        "gender": null,
        "create_time": "2020-06-11 18:17:40",
        "operate_time": "2022-02-23 21:22:23",
        "status": null
    },
    "old": {
        "nick_name": "妹妹",
        "operate_time": "2022-02-23 21:18:31"
    }
}
         */
        //todo 2.创建动态表topic_db连接kafka topic_db主题
        tableEnv.executeSql(KafkaUtil.getKafkaTopicDbDDL("dwdtradeorderdetail_220828"));




        //todo 3.过滤出订单明细表
        Table orderDetailTable = tableEnv.sqlQuery("" +
                "select \n" +
                "data['id'] id, \n" +
                "data['order_id'] order_id, \n" +
                "data['sku_id'] sku_id, \n" +
                "data['sku_name'] sku_name, \n" +
                "data['order_price'] order_price, \n" +
                "data['sku_num'] sku_num, \n" +
                "data['create_time'] create_time, \n" +
                "data['source_type'] source_type, \n" +
                "data['source_id'] source_id, \n" +
                "data['split_total_amount'] split_total_amount, \n" +
                "data['split_activity_amount'] split_activity_amount, \n" +
                "data['split_coupon_amount'] split_coupon_amount," +
                "ts,\n" +//用于后面的支付成功表lookup join，用来后面当处理时间
                "pt\n" +//将建表的处理时间拿过来，后面lookup join要用
                "from topic_db\n" +
                "where \n" +
                "`database`='gmall' \n" +
                "and `table`='order_detail'\n" +
                "and type='insert' ");


//        orderDetailTable.execute().print();
        tableEnv.createTemporaryView("order_detail", orderDetailTable);


        //todo 4.过滤出订单表
        Table orderInfo = tableEnv.sqlQuery("" +
                "select \n" +
                "data['id'] id,\n" +
                "data['consignee'] consignee,\n" +
                "data['consignee_tel'] consignee_tel,\n" +
                "data['total_amount'] total_amount,\n" +
                "data['order_status'] order_status,\n" +
                "data['user_id'] user_id,\n" +
                "data['payment_way'] payment_way,\n" +
                "data['delivery_address'] delivery_address,\n" +
                "data['order_comment'] order_comment,\n" +
                "data['out_trade_no'] out_trade_no,\n" +
                "data['trade_body'] trade_body,\n" +
                "data['create_time'] order_create_time,\n" +
                "data['operate_time'] order_operate_time,\n" +
                "data['expire_time'] order_expire_time,\n" +
                "data['process_status'] process_status,\n" +
                "data['tracking_no'] tracking_no,\n" +
                "data['parent_order_id'] parent_order_id,\n" +
                "data['province_id'] province_id,\n" +
                "data['activity_reduce_amount'] activity_reduce_amount,\n" +
                "data['coupon_reduce_amount'] coupon_reduce_amount,\n" +
                "data['original_total_amount'] original_total_amount,\n" +
                "data['feight_fee'] feight_fee,\n" +
                "data['feight_fee_reduce'] feight_fee_reduce,\n" +
                "data['refundable_time'] refundable_time\n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table`='order_info'\n" +
                "and type='insert'");
        tableEnv.createTemporaryView("order_info",orderInfo);



        //todo 5.过滤出优惠卷明细表
        Table orderDetailCoupon = tableEnv.sqlQuery("" +
                "select \n" +
                "data['id'] id,\n" +
                "data['order_id'] order_id,\n" +
                "data['order_detail_id'] order_detail_id,\n" +
                "data['coupon_id'] coupon_id,\n" +
                "data['coupon_use_id'] coupon_use_id,\n" +
                "data['sku_id'] sku_id,\n" +
                "data['create_time'] create_time\n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table`='order_detail_info'\n" +
                "and type='insert'");
        tableEnv.createTemporaryView("order_detail_coupon",orderDetailCoupon);

        //todo 6.过滤出活动明细表
        Table orderDetailActivity = tableEnv.sqlQuery("" +
                "select \n" +
                "data['id'] id,\n" +
                "data['order_id'] order_id,\n" +
                "data['order_detail_id'] order_detail_id,\n" +
                "data['activity_id'] activity_id,\n" +
                "data['activity_rule_id'] activity_rule_id,\n" +
                "data['sku_id'] sku_id,\n" +
                "data['create_time'] create_time\n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table`='order_detail_activity'\n" +
                "and type='insert'");
        tableEnv.createTemporaryView("order_detail_activity",orderDetailActivity);

        //todo 7.创建动态表base_dic连接mysql里的base_dic表
        tableEnv.executeSql(MySqlUtil.getBaseDicDDL());

        //todo 8.五表关联
        Table dwdOrderDetail = tableEnv.sqlQuery("" +
                "select \n" +
                "od.id,\n" +
                "od.order_id,\n" +
                "od.sku_id,\n" +
                "od.sku_name,\n" +
                "od.order_price,\n" +
                "od.sku_num,\n" +
                "od.create_time,\n" +
                "od.source_type,\n" +
                "od.source_id,\n" +
                "od.split_total_amount,\n" +
                "od.split_activity_amount,\n" +
                "od.split_coupon_amount,\n" +
                "od.ts," +
                "oi.total_amount,\n" +
                "oi.order_status,\n" +
                "oi.user_id,\n" +
                "oi.payment_way,\n" +
                "oi.order_create_time,\n" +
                "order_operate_time,\n" +
                "order_expire_time,\n" +
                "process_status,\n" +
                "province_id,\n" +
                "activity_reduce_amount,\n" +
                "coupon_reduce_amount,\n" +
                "original_total_amount,\n" +
                "feight_fee,\n" +
                "feight_fee_reduce,\n" +
                "refundable_time,\n" +
                "act.activity_id,\n" +
                "act.activity_rule_id,\n" +
//                "order_price,\n" +
////                "create_time,\n" +
//                "source_type,\n" +
//                "act.source_id,\n" +
//                "act.split_total_amount,\n" +
//                "act.split_activity_amount,\n" +
//                "act.split_coupon_amount,\n" +
                "dic.dic_name source_type_name\n" +
                "from order_detail od\n" +
                "join order_info oi on od.order_id=oi.id\n" +
                "left join order_detail_coupon cou on od.id=cou.order_detail_id\n" +
                "left join order_detail_activity act on od.id=act.order_detail_id\n" +
                "join base_dic for system_time as of od.pt as dic on od.source_type=dic.dic_code\t");
        tableEnv.createTemporaryView("result_table",dwdOrderDetail);


        //todo 9.创建动态表dwd_order_detail连接kafka dwd_trade_order_detail主题
        tableEnv.executeSql("create table dwd_order_detail(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "sku_id string,\n" +
                "sku_name string,\n" +
                "order_price string,\n" +
                "sku_num string,\n" +
                "create_time string,\n" +
                "source_type string,\n" +
                "source_id string,\n" +
                "split_total_amount string,\n" +
                "split_activity_amount string,\n" +
                "split_coupon_amount string,\n" +
                "ts string," +
                "total_amount string,\n" +
                "order_status string,\n" +
                "user_id string,\n" +
                "payment_way string,\n" +
                "order_create_time string,\n" +
                "order_operate_time string,\n" +
                "order_expire_time string,\n" +
                "process_status string,\n" +
                "province_id string,\n" +
                "activity_reduce_amount string,\n" +
                "coupon_reduce_amount string,\n" +
                "original_total_amount string,\n" +
                "feight_fee string,\n" +
                "feight_fee_reduce string,\n" +
                "refundable_time string,\n" +
                "activity_id string,\n" +
                "activity_rule_id string,\n" +
//                "order_price string,\n" +
//                "source_type string,\n" +
//                "source_id string,\n" +
//                "split_total_amount string,\n" +
//                "split_activity_amount string,\n" +
//                "split_coupon_amount string,\n" +
                "source_type_name string,\n" +
                "primary key (id) not enforced)"+KafkaUtil.getUpsertKafkaDDL("dwd_trade_order_detail"));
        //todo 10.将关联后的结果写到dwd_order_detail
        tableEnv.executeSql("insert into dwd_order_detail select * from result_table");


    }
}
